package com.sscw.tcp.tcp;

import com.sscw.tcp.dao.GConfDeviceDAO;
import com.sscw.tcp.constant.MQTTTopicConstant;
import com.sscw.tcp.constant.TcpCommandCodeConstant;
import com.sscw.tcp.entity.GConfDevicePO;
import com.sscw.tcp.mqtt.MQTTProducer;
import com.sscw.tcp.utils.TcpUtils;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;

import java.io.InputStream;
import java.net.Socket;
import java.util.Objects;

@Slf4j
public class TcpListenerThread extends Thread {

    private String deviceNo;

    private Socket socket;

    private MQTTProducer messageProducer;

    private GConfDeviceDAO gConfDeviceDAO;

    private RedissonClient redissonClient;

    public TcpListenerThread(Socket socket, MQTTProducer messageProducer, GConfDeviceDAO gConfDeviceDAO, RedissonClient redissonClient) {
        this.socket = socket;
        this.messageProducer = messageProducer;
        this.gConfDeviceDAO = gConfDeviceDAO;
        this.redissonClient = redissonClient;
    }

    @SneakyThrows
    @Override
    public void run() {
        while (!TcpUtils.clientIsClose(socket)) {
            try {
                InputStream is = socket.getInputStream();
                StringBuilder sb = new StringBuilder();
                byte[] bytes = new byte[1024];
                int len = is.read(bytes);
                for (int i = 0; i < len; i++) {
                    String hexString = String.format("%02X", bytes[i]);
                    sb.append(hexString);
                }
                String message = sb.toString();
                //判断是否是结算心跳
                if (message.startsWith("8A0912030100000000001F")) {
                    if (StringUtils.isNotBlank(deviceNo)) {
                        RBucket<String> bucket = redissonClient.getBucket(deviceNo);
                        if (Objects.isNull(bucket) || StringUtils.isBlank(bucket.get())) {
                            bucket.set(deviceNo);
                        }
                    }
                    continue;
                }
                if (message.startsWith("8A0912030000000000001E")) {
                    if (StringUtils.isNotBlank(deviceNo)) {
                        RBucket<String> bucket = redissonClient.getBucket(deviceNo);
                        if (Objects.nonNull(bucket) && StringUtils.isNotBlank(bucket.get())) {
                            TcpUtils.sendMessage(socket, TcpCommandCodeConstant.SETTLE_APPLY);
                            bucket.delete();
                        }
                    }
                    continue;
                }
                log.info("接收到设备的消息,{}", message);
                if (StringUtils.isNotBlank(message) && message.startsWith("8A0A0102")) {
                    String substring = message.substring(8);
                    String deviceNo = substring.substring(0, substring.length() - 4);
                    this.deviceNo = deviceNo;
                    log.info("连接设备编码为：{}", deviceNo);
                    TcpUtils.addSocket(deviceNo, socket);
                    GConfDevicePO devicePO = gConfDeviceDAO.findOne(GConfDevicePO::getSerialNumber, deviceNo);
                    if (Objects.isNull(devicePO)) {
                        GConfDevicePO confDevicePO = gConfDeviceDAO.findOneDeviceNoIsNull();
                        if (Objects.nonNull(confDevicePO)) {
                            confDevicePO.setSerialNumber(deviceNo);
                            gConfDeviceDAO.updateById(confDevicePO);
                        }
                    }
                } else if (StringUtils.isNotBlank(message) && message.startsWith("8A070A03")) {
                    messageProducer.sendIoMessage(MQTTTopicConstant.IO_FIRST_TOPIC + MQTTTopicConstant.SETTLE_MACHINE, deviceNo + "_" + message, 1);
                    TcpUtils.sendMessage(socket, TcpCommandCodeConstant.STOP);
                }
            } catch (Exception e) {
                log.error("处理串口数据异常：{}", e);
            }
        }
    }
}
